package net.bwie.realtime.vehicle.dwd.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

/**
 * @author ASUS
 */
public class OverSpeedRichFilterFunction<String, Boolean> extends RichFilterFunction<String> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public boolean filter(String string) throws Exception {
        try {
            // 1. 解析JSON（捕获解析异常，避免单个坏数据导致任务失败）
            JSONObject jsonObj = JSON.parseObject(java.lang.String.valueOf(string));
            if (jsonObj == null) {
                return false;
            }
            // 2. 提取实时速度（必须从当前JSON中获取，而非new对象）
            Integer realSpeed = jsonObj.getInteger("speed");
            // 3. 提取限速（processJson已确保speedLimit不为null，但仍做兜底）
            Integer speedLimit = jsonObj.getInteger("speedLimit");
            // 4. 校验字段有效性（避免null参与计算）
            if (realSpeed == null) {
                return false;
            }
            if (speedLimit == null) {
                speedLimit = 60;
            }
            // 5. 超速判断（实时速度 > 限速*1.1）
            return realSpeed > speedLimit * 1.1;
        } catch (Exception e) {
            // 捕获所有异常（如JSON格式错误），避免任务崩溃
            return false;
        }
    }
    @Override
    public void close() throws Exception {
        super.close();
    }
}
